package com.hopu.user.era;

import com.alibaba.fastjson.JSONObject;

import com.hopu.bean.era.EraBean;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import scala.Tuple2;

import java.io.IOException;

public class EraOperate {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("eraOperate").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        JavaPairRDD<EraBean, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String era = json.getJSONObject("birthday").getString("year").substring(2, 3) + "0";
            String event_type = json.getString("event_type");
            return new Tuple2<>(new EraBean(era, event_type), 1);
        });
        /*
        pairRDD：
        (EraBean(era=60, event_type=view, viewCount=0, cartCount=0, parchaseCount=0, totalCount=0),1)
        (EraBean(era=50, event_type=cart, viewCount=0, cartCount=0, parchaseCount=0, totalCount=0),1)
        (EraBean(era=00, event_type=view, viewCount=0, cartCount=0, parchaseCount=0, totalCount=0),1)
        (EraBean(era=80, event_type=parchase, viewCount=0, cartCount=0, parchaseCount=0, totalCount=0),1)
         */

        //过滤出各年代的cart类型的操作
        String[] eras = {"50", "60", "70", "80", "90", "00", "10"};
        for (String era: eras) {
            filterEraAndType(pairRDD, era);
        }
    }

    public static void filterEraAndType(JavaPairRDD<EraBean, Integer> pairRDD, String era) {
        pairRDD = pairRDD.filter(t -> t._1.getEra().equals(era));//匹配出当前所有 为 era 年代的 pairRDD

        EraBean view = getResult(pairRDD, "view");
        EraBean cart = getResult(pairRDD, "cart");
        EraBean parchase = getResult(pairRDD, "parchase");

//        JavaPairRDD<EraBean, Integer> viewRDD = pairRDD.filter(t -> t._1.getEvent_type().equals("view"));
//        JavaPairRDD<EraBean, Integer> cartRDD = pairRDD.filter(t -> t._1.getEvent_type().equals("cart"));
//        JavaPairRDD<EraBean, Integer> parchaseRDD = pairRDD.filter(t -> t._1.getEvent_type().equals("parchase"));
//
//        //leftOuterJoin：左外连接。根据Tuple2的第一个参数来进行连接，Tuple2的第二个参数进行连接，存放到Optional类的对象中
////        JavaPairRDD<EraBean, Tuple2<Integer, Optional<Integer>>> viewAndCartRDD = viewRDD.leftOuterJoin(cartRDD);
//
//        JavaPairRDD<EraBean, Iterable<Integer>> viewGroup = viewRDD.groupByKey();
//        EraBean view = viewGroup.map(t -> {
//            int count = 0;
//            for (int v : t._2) {
//                count += v;
//            }
//            t._1.setViewCount(count);
//            return t._1;
//        }).first();
//
//        JavaPairRDD<EraBean, Iterable<Integer>> cartGroup = cartRDD.groupByKey();
//        EraBean cart = cartGroup.map(t -> {
//            int count = 0;
//            for (int v : t._2) {
//                count += v;
//            }
//            t._1.setCartCount(count);
//            return t._1;
//        }).first();
//
//        JavaPairRDD<EraBean, Iterable<Integer>> parchaseGroup = parchaseRDD.groupByKey();
//        EraBean parchase = parchaseGroup.map(t -> {
//            int count = 0;
//            for (int v : t._2) {
//                count += v;
//            }
//            t._1.setParchaseCount(count);
//            return t._1;
//        }).first();

        view.setCartCount(cart.getCartCount());
        view.setParchaseCount(parchase.getParchaseCount());
        view.setTotalCount(view.getViewCount()+view.getCartCount()+view.getParchaseCount());
        System.out.println(view);

        String[] columns = {"era", "viewCount", "cartCount", "parchaseCount", "totalCount"};
        String[] values = {view.getEra(), view.getViewCount()+"", view.getCartCount()+"", view.getParchaseCount()+"",
                view.getTotalCount()+""};
        try {
            HBaseUtils.putOneRowToHbase("era", view.getEra(), "totalOperate", columns, values);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static EraBean getResult(JavaPairRDD<EraBean, Integer> pairRDD, String event_type) {
        pairRDD = pairRDD.filter(t -> t._1.getEvent_type().equals(event_type));
        JavaPairRDD<EraBean, Iterable<Integer>> group = pairRDD.groupByKey();

        EraBean result = group.map(t -> {
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            if (event_type.equals("view")) {
                t._1.setViewCount(count);
            } else if (event_type.equals("cart")) {
                t._1.setCartCount(count);
            } else {
                t._1.setParchaseCount(count);
            }
            return t._1;
        }).first();
        return result;
    }
}
